package cn.jingyuan.swan.example.nio.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;

public class ServerReactorTask implements Runnable {

    private Selector selector;

    private volatile boolean runningFlag = true;

    public ServerReactorTask() {
        init();
    }

    private void init() {
        try {
            // 第一步：打开 ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父管道
            ServerSocketChannel serverChannel = ServerSocketChannel.open();

            // 第二步：病毒监听端口，设置连接为非阻塞模式
            serverChannel.configureBlocking(false);
            serverChannel.bind(new InetSocketAddress(9090), 2);

            // 第三步：创建 Reactor 线程，创建多路复用器并启动线程
            selector = Selector.open();

            // 第四步：将ServerSocketChannel注册到Reactor线程的多路复用器Selector上，监听Accept事件
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            System.out.println("server started...");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        // 第五步：在 run 方法中无限循环体内轮询准备就绪的 Key
        while (runningFlag) {
            try {
                int selectCount = selector.select(1000);
                if (0 == selectCount) {
                    continue;
                }

                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> keysIterator = selectionKeys.iterator();

                while (keysIterator.hasNext()) {
                    SelectionKey key = keysIterator.next();
                    keysIterator.remove();

                    if (Boolean.FALSE.equals(key.isValid())) {
                        continue;
                    }

                    // 处理新接入的请求消息
                    if (key.isAcceptable()) {
                        // 第六步：多路复用器监听到有新的客户端接入，处理新的接入请求，完成 TCP 三次握手，建立物理链路
                        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();

                        SocketChannel clientChannel = serverChannel.accept();

                        // 第七步：设置客户端链路为非阻塞模式
                        clientChannel.configureBlocking(false);
                        // clientChannel.socket().setReuseAddress(true);


                        // 第八步：将新接入的客户端连接注册到 Reactor 线程的多路复用器上，监听读操作，读取客户端发送的网络消息
                        clientChannel.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) {
                        // 第九步：异步读取客户端请求消息到缓存区
                        SocketChannel clientChannel = (SocketChannel) key.channel();

                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);

                        int read = clientChannel.read(readBuffer);

                        // 第十步：对 ByteBuffer 进行编解码，如果有半包消息指针 reset，继续读取后续的报文
                        if (read > 0) {
                            readBuffer.flip();

                            byte[] readDataBytes = new byte[readBuffer.remaining()];

                            readBuffer.get(readDataBytes);

                            String body = new String(readDataBytes, StandardCharsets.UTF_8);

                            System.out.println("The server receive: " + body);

                            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
                                ? new java.util.Date(System.currentTimeMillis()).toString()
                                : "BAD ORDER";


                            // 写应答
                            byte[] writeDataBytes = currentTime.getBytes(StandardCharsets.UTF_8);

                            ByteBuffer writeBuffer = ByteBuffer.allocate(writeDataBytes.length);
                            writeBuffer.put(writeDataBytes);
                            writeBuffer.flip();

                            clientChannel.write(writeBuffer);
                        } else if (read < 0) {
                            // 对端链路关闭
                            key.cancel();
                            clientChannel.close();
                        } else {
                            // 读到 0 字节，忽略
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    public void stopServer() {
        runningFlag = false;
    }

}
